package cn.getech.data.development.utils;

import cn.getech.data.development.constant.StorageEngineEnum;
import cn.getech.data.development.entity.TableInfo;
import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSONObject;
import com.xxl.job.core.log.XxlJobLogger;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class HiveTableUtil {


    private Logger log = LoggerFactory.getLogger(HiveTableUtil.class);
    private Connection con;
    private Statement sta;
    private PreparedStatement psta;

    public HiveTableUtil(String serverUrl, String userName, String password) {
        try {
            Class.forName("org.apache.hive.jdbc.HiveDriver");
            DriverManager.setLoginTimeout(50);
            con = DriverManager.getConnection(
                    serverUrl, userName, password);
            //con = druidUtil.getHiveConnection(serverUrl, userName, password);
        } catch (Exception e) {
            XxlJobLogger.log("xxllog获取数据库连接异常{}",e.getMessage());
            XxlJobLogger.log(e);
            log.error(e.getMessage());
        }
    }

    public HiveTableUtil(String serverUrl, String dbName, String userName, String password) {
        try {

            if (serverUrl.contains("21050")) {//为kudu 连接方式
                String connectProp = ";auth=noSasl";
                if (dbName != null) {
                    serverUrl = serverUrl + dbName+connectProp;
                }else{
                    serverUrl = serverUrl +connectProp;
                }
            }else{
                if (dbName != null) {
                    serverUrl = serverUrl + dbName;
                }
            }

            Class.forName("org.apache.hive.jdbc.HiveDriver");
            DriverManager.setLoginTimeout(50);
            con = DriverManager.getConnection(
                    serverUrl, userName, password);
            // con = druidUtil.getHiveConnection(serverUrl, userName, password);
        } catch (Exception e) {
            e.printStackTrace();
            XxlJobLogger.log("xxllog获取数据库连接异常{}",e.getMessage());
            XxlJobLogger.log(e);
            log.error(e.getMessage());
        }
    }

    public HiveTableUtil(String serverUrl, String queen, String dbName, String userName, String password) {
        try {
            if (serverUrl.contains("21050") && dbName != null) {//为kudu 连接方式
                dbName = dbName + ";auth=noSasl";
            }
            if (dbName != null) {
                serverUrl = serverUrl  +dbName;
            }
            serverUrl=serverUrl+"?mapreduce.job.queuename="+queen;
            Class.forName("org.apache.hive.jdbc.HiveDriver");
            DriverManager.setLoginTimeout(50);
            con = DriverManager.getConnection(
                    serverUrl, userName, password);
            // con = druidUtil.getHiveConnection(serverUrl, userName, password);
        } catch (Exception e) {
            XxlJobLogger.log("xxllog获取数据库连接异常{}",e.getMessage());
            XxlJobLogger.log(e);
            log.error(e.getMessage());
        }
    }

    public Long dataPreviewRowNum(TableInfo tableInfo) {
        Long totalCount = 0L;
        try {
            ResultSet resultNum = getResultSet(String.format("select count(1) as count from %s", tableInfo.getTableName()));
            while (resultNum.next()) {
                totalCount += resultNum.getLong("count");
            }
        } catch (SQLException e) {
            log.error(e.getMessage());
        } finally {
            close();
        }
        return totalCount;
    }

    public Long dataPreviewTotalCount(String sql) {
        Long totalCount = 0L;
        try {
            ResultSet resultNum = getResultSet(String.format("select count(1) as count from (%s )temp", sql));
            while (resultNum.next()) {
                totalCount += resultNum.getLong("count");
            }
        } catch (SQLException e) {
            log.error(e.getMessage());
        } finally {
//            close();
        }
        return totalCount;
    }

    public void analyzeTable(TableInfo tableInfo, List<String> partitions) throws SQLException {
        if (con != null) {
            sta = con.createStatement();
            if (CollectionUtil.isNotEmpty(partitions)) {
                if (partitions.size() == 0) {
                    return;
                }
                for (String p : partitions) {
                    sta.execute(String.format("ANALYZE TABLE %s.%s PARTITION(%s) COMPUTE STATISTICS", tableInfo.getDbName(), tableInfo.getTableName(), p));
                }
            } else {
                sta.execute(String.format("ANALYZE TABLE %s.%s  COMPUTE STATISTICS", tableInfo.getDbName(), tableInfo.getTableName()));
            }
            log.debug(String.format("成功统计%s.%s的数据条数", tableInfo.getDbName(), tableInfo.getTableName()));
            close();
        } else{
            throw new SQLException("hive连接失败,请稍后尝试");
        }
    }

    /**
     * 统计分区表，逐个分区统计
     *
     * @param tableInfo
     * @param partition
     * @throws SQLException
     */
    public void analyzeTableSinglePartition(TableInfo tableInfo, String partition) throws SQLException {
        StringBuilder sql = new StringBuilder();
        String analyzeTable = String.format("ANALYZE TABLE %s.%s ", tableInfo.getDbName(), tableInfo.getTableName());
        if (StringUtils.isNotEmpty(partition)) {
            sql.append(analyzeTable).append(String.format("PARTITION(%s) COMPUTE STATISTICS", partition));
        } else {
            sql.append(analyzeTable).append("COMPUTE STATISTICS");
        }
        if (StringUtils.isNotBlank(sql.toString())) {
            log.debug(String.format("统计%s.%s的数据条数的sql:%s", tableInfo.getDbName(), tableInfo.getTableName(), sql.toString()));
            execute(sql.toString());
            log.debug(String.format("成功统计%s.%s的数据条数", tableInfo.getDbName(), tableInfo.getTableName()));
        }
    }


    public List tableInfoDataPreview(TableInfo tableInfo, List<String> columnList, Integer page, Integer limit) {
        List<JSONObject> data = new ArrayList<>();
        Integer start = (page - 1) * limit;

        if (CollectionUtil.isEmpty(columnList)) {
            return data;
        }
        try {
            String sql = "select ";
            for (String columns : columnList) {
                sql = sql + columns + " ,";
            }
            sql = sql.substring(0, sql.length() - 1);
            if (StorageEngineEnum.KUDU.getCode().equals(tableInfo.getStoreEngine())) {
                sql = sql + " from  %s.%s limit  %d offset %d";
                sql = String.format(sql, tableInfo.getDbName(), tableInfo.getTableName(), limit,start);
            } else {
                sql = sql + " from  %s.%s limit %d, %d";
                sql = String.format(sql, tableInfo.getDbName(), tableInfo.getTableName(), start, limit);
            }
            ResultSet result = getResultSet(sql);
            ResultSetMetaData tsmd = result.getMetaData();
            while (result.next()) {
                JSONObject object = new JSONObject();
                for (int i = 1; i < tsmd.getColumnCount() + 1; i++) {
                    String titleName = tsmd.getColumnName(i).replaceAll(String.format("%s.", tableInfo.getTableName()), "");
                    for (String c : columnList
                    ) {
                        if (c.equalsIgnoreCase(titleName)) {
                            object.put(c, result.getObject(i));

                            break;
                        }
                    }

                }
                data.add(object);
            }
        } catch (SQLException e) {
            log.error(e.getMessage());
        } finally {
            close();
        }
        return data;
    }


    public Map<String, Object> tableInfoDataPreviewPage(TableInfo tableInfo, List<String> columnList, Integer page, Integer limit) {
        Map<String, Object> resMap = new HashMap<>();
        List<JSONObject> data = new ArrayList<>();
        Integer start = (page - 1) * limit;

        if (CollectionUtil.isEmpty(columnList)) {
            return resMap;
        }
        try {
            String sql = "select ";
            for (String columns : columnList) {
                sql = sql + columns + " ,";
            }
            sql = sql.substring(0, sql.length() - 1) + " from %s.%s ";
            sql = String.format(sql, tableInfo.getDbName(), tableInfo.getTableName());

            //获取总数
//            Long totalCount = dataPreviewTotalCount(sql);
            Long totalCount = 1111110L;

            resMap.put("totalCount",totalCount);

            if (StorageEngineEnum.KUDU.getCode().equals(tableInfo.getStoreEngine())) {
                sql = sql + " limit  %d";
                sql = String.format(sql, limit);
            } else {
                sql = sql + " limit %d, %d";
                sql = String.format(sql, start, limit);
            }
            ResultSet result = getResultSet(sql);
            ResultSetMetaData tsmd = result.getMetaData();
            while (result.next()) {
                JSONObject object = new JSONObject();
                for (int i = 1; i < tsmd.getColumnCount() + 1; i++) {
                    String titleName = tsmd.getColumnName(i).replaceAll(String.format("%s.", tableInfo.getTableName()), "");
                    for (String c : columnList
                    ) {
                        if (c.equalsIgnoreCase(titleName)) {
                            object.put(c, result.getObject(i));

                            break;
                        }
                    }

                }
                data.add(object);
            }
        } catch (SQLException e) {
            log.error(e.getMessage());
        } finally {
            close();
        }
        resMap.put("data",data);
        return resMap;
    }

    public Map<String, Object> dataPreview(Map<String, Object> params) {
        String tableName = params.get("tableName").toString();
        List<String> columnList = new ArrayList<>();
        List<JSONObject> data = new ArrayList<>();
        Map<String, Object> dataMap = new HashMap<>();
        try {
            String sql = String.format("select * from  %s", tableName);
            if (params.get("sql") != null) {
                sql = params.get("sql").toString();
            }
            //导出不用分页
            if (null != params.get("limit") && null != params.get("page")) {
                Integer pageSize = Integer.parseInt(params.get("limit").toString());
                Integer currPage = Integer.parseInt(params.get("page").toString());
                Integer start = (currPage - 1) * pageSize;
                if (params.get("searchEngine") != null) {//kudu表
                    sql = sql + String.format(" limit  %d", pageSize);
                }else{
                    sql = sql + String.format(" limit %d, %d", start, pageSize);
                }
            }
            ResultSet result = getResultSet(sql);
            ResultSetMetaData tsmd = result.getMetaData();
            for (int i = 1; i < tsmd.getColumnCount() + 1; i++) {
                columnList.add(tsmd.getColumnName(i).replaceAll(String.format("%s.", tableName), ""));
            }
            dataMap.put("columnList", columnList);
            while (result.next()) {
                JSONObject object = new JSONObject();
                for (int i = 1; i < tsmd.getColumnCount() + 1; i++) {
                    object.put(tsmd.getColumnName(i).replaceAll(String.format("%s.", tableName), ""), result.getObject(i));
                }
                data.add(object);
            }
            dataMap.put("data", data);
        } catch (SQLException e) {
            log.error("sql执行错误",e);
        } finally {
            close();
        }
        return dataMap;
    }

    public static String genTableLocation(String dbName, String tableName) {
        String tabUrl = "default".equals(dbName) ? tableName : dbName + ".db/" + tableName;
        return "/user/hive/warehouse/" + tabUrl;
    }


    public void dropTable(String tableName) throws SQLException {
        if (con != null) {
            sta = con.createStatement();
            sta.execute(String.format("DROP TABLE IF EXISTS %s", tableName));
        }else{
            throw new SQLException("连接hive失败,连接信息为空");
        }
    }


    public ResultSet getResultSet(String sql) throws SQLException {
        if (con != null) {
            psta = con.prepareStatement(sql);
            return psta.executeQuery();
        }else{
            throw new SQLException("hive连接失败,请稍后尝试");
        }
    }

    public void execute(String script) throws SQLException {
        if (con != null) {
            sta = con.createStatement();
            sta.execute(script);
        }else{
            throw new SQLException("hive连接失败,请稍后尝试");
        }
    }

    public void close() {
        try {
            if (sta != null) {
                sta.close();
            }
            if (psta != null) {
                psta.close();
            }
            if (con != null) {
                con.close();
            }
        } catch (Exception e) {
            log.error("关闭异常:{}", e.getMessage());
        }
    }

    public void checkHiveSql(String sql) {
        //暂时不做sql计算
//        ParseDriver pd = new ParseDriver();
//        try{
//            ASTNode tree = pd.parse(sql);
//        } catch (Exception e) {
//            throw new RRException("sql语句：'"+sql+"'存在语法错误");
//        }
    }



    public void repairTable(String tableName) throws SQLException {
        //用权限的hiveurl会报错，报修改表失败 set 会报错的问题（已经改完了）
        String sql = "set hive.msck.path.validation = skip ";
        if (con != null) {
            sta = con.createStatement();
            sta.execute(sql);
            sta.execute(String.format("msck repair table %s", tableName));
        }else{
            throw new SQLException("hive连接失败,请稍后尝试");
        }

    }


    /***
     * 慎重修改，如果修改，则还得重新发布job-center服务
     * @param tableName
     * @return
     */
    public Long dataPreviewRowNum(String tableName) {
        Long totalCount = -1L;
        try {
            ResultSet resultNum = getResultSet(String.format("select sum(1) as num from %s", tableName));
            while (resultNum.next()) {
                totalCount += resultNum.getLong("num");
            }
        } catch (SQLException e) {
            log.error(e.getMessage());
        } finally {
            close();
        }
        return totalCount;
    }
}
